package cn.me.alphamq.client.consumer;

import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.constant.GlobalConstant;
import cn.me.alphamq.common.entity.req.*;
import cn.me.alphamq.common.entity.resp.*;
import cn.me.alphamq.common.enumeration.*;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.remote.DefaultClient;
import cn.me.alphamq.common.remote.NettyClient;
import cn.me.alphamq.common.util.AssertUtil;
import cn.me.alphamq.common.util.TopicNameUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 基础的消费者类
 *
 * @author f
 */
public class DefaultConsumer extends DefaultClient {

    private static final Logger log = LoggerFactory.getLogger(DefaultConsumer.class);

    /**
     * 消费者消费的主题
     */
    private final String topic;

    /**
     * 消费者重试消费的主题
     */
    private final String retryTopic;

    /**
     * 消费者的消费者组名称
     */
    private final String consumerGroup;

    /**
     * 消费队列的 id
     */
    private final int queueId;

    /**
     * 消费者一次拉取的消息数量
     */
    private final int batchMsgNum;

    /**
     * 消费最大重试次数，默认为 9 次，取值范围是 0~9
     */
    private final int maxRetryTimes;

    /**
     * 消息消费的监听器
     */
    private final MessageConsumeListener messageConsumeListener;

    // 开启消费消息的任务
    private void startConsumeMsgTask(String topicName) {
        new Thread(() -> {
            while (true) {
                try {
                    NettyClient brokerClient = pickBrokerClient();

                    // 获取消息
                    List<Message> messageList = fetchMessage(brokerClient, topicName).getMessageList();
                    if (messageList.isEmpty()) {
                        TimeUnit.MILLISECONDS.sleep(GlobalConstant.NON_EXIST_DATA_TIME_STEP);
                        continue;
                    }

                    // 进行消费
                    ConsumeResult consumeResult = messageConsumeListener.consume(messageList);
                    if (ConsumeResult.CONSUME_SUCCESS == consumeResult) {
                        // 如果消费成功，则返回 ack 消息
                        sendAckMsg(brokerClient, topicName, messageList.size());
                    } else if (ConsumeResult.CONSUME_LATER == consumeResult) {
                        // 如果消费失败，则返回 重试 消息
                        sendRetryMsg(brokerClient, topicName, messageList);
                    }

                    TimeUnit.MILLISECONDS.sleep(GlobalConstant.EXIST_DATA_TIME_STEP);
                } catch (InterruptedException e) {
                    log.error("", e);
                }
            }
        }, "consume-msg-thread").start();
    }

    // 给 broker 发送 重试 消息
    private void sendRetryMsg(NettyClient brokerClient, String topicName, List<Message> messageList) {
        TcpMsg resp = brokerClient.sendMsgSync(EventCodeEnum.RETRY_CONSUME_MSG_REQ,
                new RetryConsumeMsgReq(topicName, consumerGroup, queueId, maxRetryTimes, messageList));
        RetryConsumeMsgResp retryConsumeMsgResp = JSON.parseObject(resp.getBody(), RetryConsumeMsgResp.class);
        if (AckStatusEnum.FAIL.getCode() == retryConsumeMsgResp.getRetryStatus()) {
            log.info("retry fail!");
        }
    }

    // 给 broker 发送 ack 消息
    private void sendAckMsg(NettyClient brokerClient, String topicName, int consumeMsgNum) {
        TcpMsg resp = brokerClient.sendMsgSync(EventCodeEnum.CONSUME_SUCCESS_REQ,
                new ConsumeSuccessReq(topicName, consumerGroup, queueId, consumeMsgNum));
        ConsumeSuccessResp consumeSuccessResp = JSON.parseObject(resp.getBody(), ConsumeSuccessResp.class);
        if (AckStatusEnum.FAIL.getCode() == consumeSuccessResp.getAckStatus()) {
            log.error("consume ack fail!");
        }
    }

    // 给 broker 发送创建重试 topic 的消息
    private void sendCreateRetryTopicMsg(NettyClient brokerClient) {
        brokerClient.sendMsgAsync(EventCodeEnum.CREATE_TOPIC_REQ, new CreateTopicReq(
                TopicNameUtil.getRetryTopicName(consumerGroup),
                GlobalConstant.DEFAULT_RETRY_QUEUE_SIZE
        ));
    }

    // 从 broker 中获取消费的消息
    private ConsumeMsgResp fetchMessage(NettyClient brokerClient, String topicName) {
        ConsumeMsgReq consumeMsgReq = new ConsumeMsgReq(topicName, consumerGroup, queueId, batchMsgNum);
        TcpMsg resp = brokerClient.sendMsgSync(EventCodeEnum.CONSUME_MSG_REQ, consumeMsgReq);
        if (resp == null) {
            ConsumeMsgResp consumeMsgResp = new ConsumeMsgResp();
            consumeMsgResp.setMessageList(new ArrayList<>());
            return consumeMsgResp;
        }
        return JSON.parseObject(resp.getBody(), ConsumeMsgResp.class);
    }

    public DefaultConsumer(String nameServerIp, int nameServerPort,
                           String nameServerUsername, String nameServerPassword,
                           String topic, String consumerGroup, int queueId,
                           int batchMsgNum, String brokerRole, String brokerGroup,
                           MessageConsumeListener messageConsumeListener) {
        this(nameServerIp, nameServerPort, nameServerUsername, nameServerPassword,
                topic, consumerGroup, queueId, batchMsgNum, 9, brokerRole, brokerGroup, messageConsumeListener);
    }

    public DefaultConsumer(String nameServerIp, int nameServerPort, String username, String password,
                           String topic, String consumerGroup, int queueId,
                           int batchMsgNum, int maxRetryTimes, String brokerRole, String brokerGroup,
                           MessageConsumeListener messageConsumeListener) {
        super(brokerRole, brokerGroup, nameServerIp, nameServerPort, "default-consumer-nameserver-client");

        AssertUtil.isInClosedRange(maxRetryTimes, 0, 9, "retryTimes only be assigned with 0~9");
        this.topic = topic;
        this.retryTopic = TopicNameUtil.getRetryTopicName(consumerGroup);
        this.consumerGroup = consumerGroup;
        this.queueId = queueId;
        this.batchMsgNum = batchMsgNum;
        this.maxRetryTimes = maxRetryTimes;
        this.messageConsumeListener = messageConsumeListener;

        super.start(username, password, ServiceInstanceTypeEnum.PRODUCER);
    }

    @Override
    protected String genBrokerClientName(String brokerAddress) {
        return "default-consumer-broker-client-on[" + brokerAddress + "]";
    }

    @Override
    protected void afterStartCallback() {
        // 创建一个待消费 topic 的重试 topic，同时消费这两个 topic
        sendCreateRetryTopicMsg(pickBrokerClient());
        startConsumeMsgTask(topic);
        startConsumeMsgTask(retryTopic);

        // 暂停主线程，让 startConsumeMsgTask() 里面的子线程持续运行
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * TODO 把 CountDownLatch 换成更优雅的方式
     */
    private final CountDownLatch countDownLatch = new CountDownLatch(1);
}
